# -*- coding: utf-8 -*- 
"""
Project: automation-api-project
Creator: jinsheng
Create time: 2021-11-12 11:29
IDE: PyCharm
Introduction:
"""

import json
from datetime import datetime

from kafka import KafkaProducer, KafkaConsumer
from pykafka import KafkaClient


class OperateKafka:
    def __init__(self, version=17):
        if version == 17:
            self.bootstrap_servers = ['10.0.17.17:9091', '10.0.17.17:9092', '10.0.17.17:9093']
            self.hosts = '10.0.17.17:9092'
        elif version == 221:
            self.bootstrap_servers = ['192.168.11.85:9091', '192.168.11.85:9092', '192.168.11.85:9093']
            self.hosts = '192.168.11.85:2181'
        elif version == 156:
            self.bootstrap_servers = ['cdh01:9092', 'cdh02:9092', 'cdh03:9092']
            self.hosts = '117.88.45.156:2181'
        elif version == 'dxy':
            self.bootstrap_servers = ['192.168.0.93:9091', '192.168.0.59:9091', '192.168.0.86:9091']
            # self.hosts = '117.88.45.156:2181'
        self.topic = None

    def send_message(self, topic, message):
        """
        向kafka指定topic发送消息
        @param topic: 指定的topic
        @param message: 发送的消息体(字典形式或字符串形式)
        @return: None
        """
        producer = KafkaProducer(bootstrap_servers=self.bootstrap_servers)
        send_message = None
        try:
            if isinstance(message, dict) or isinstance(message, list):
                send_message = json.dumps(message).encode()
            elif isinstance(message, str):
                send_message = message.encode()
            producer.send(topic, send_message)
        except Exception as e:
            print(e)
        finally:
            producer.close()

    def read_message(self, topic):
        """
        读取topic中的offset和value
        @param topic: 需要读取的topic
        @return: offset值和对应的value值
        """
        res_list = {}
        client = KafkaClient(hosts=self.hosts)
        topic_read = client.topics[topic]
        consumer = topic_read.get_simple_consumer(reset_offset_on_start=True)
        for message in consumer:
            if message:
                res_list[message.offset] = message.value.decode()
        return res_list

    def consumer_demo(self, topic):
        time_start = None
        count = 0
        cs = KafkaConsumer(topic, bootstrap_servers=self.bootstrap_servers)
        for message in cs:
            # res_str = message.value.decode()
            # res2 = res_str.replace('true', 'True')
            # res3 = res2.replace('false', 'False')
            # res_list = eval(res3)
            # try:
            #     response_id = res_list[0]['responseId']
            #     if 202300001 <= int(response_id[:9]) < 202330001:
            #         count += 1
            #         if count == 1:
            #             time_start = datetime.fromtimestamp(message.timestamp / 1000)
            #             print(f'第一条kafka消息入库的时间为 {time_start}')
            #     if count % 1000 == 0:
            #         time_end = datetime.fromtimestamp(message.timestamp / 1000)
            #         time_res = time_end - time_start
            #         print(f'增加了{count}个，用时{time_res}秒')
            #     if count == 30000:
            #         print('上报结束')
            #         break
            # except Exception as e:
            #     pass
            res_key = message.key.decode()
            try:
                if 202300001 <= int(res_key) < 202330001:
                    count += 1
                else:
                    pass
                if count == 1:
                    time_start = datetime.fromtimestamp(message.timestamp / 1000)
                    print(f'第一条kafka消息入库的时间为 {time_start}')
                if count % 500 == 0:
                    time_end = datetime.fromtimestamp(message.timestamp / 1000)
                    time_res = time_end - time_start
                    print(f'增加了{count}个，用时{time_res}秒')
            except Exception as e:
                # print(e)
                continue

    def consumer_demo_2(self, topic):
        time_start = None
        count = 0
        tmp_list = []
        cs = KafkaConsumer(topic, bootstrap_servers=self.bootstrap_servers)
        for message in cs:
            # res_str = message.value.decode()
            # res2 = res_str.replace('true', 'True')
            # res3 = res2.replace('false', 'False')
            # res_list = eval(res3)
            # try:
            #     response_id = res_list[0]['responseId']
            #     if 202300001 <= int(response_id[:9]) < 202330001:
            #         count += 1
            #         if count == 1:
            #             time_start = datetime.fromtimestamp(message.timestamp / 1000)
            #             print(f'第一条kafka消息入库的时间为 {time_start}')
            #     if count % 1000 == 0:
            #         time_end = datetime.fromtimestamp(message.timestamp / 1000)
            #         time_res = time_end - time_start
            #         print(f'增加了{count}个，用时{time_res}秒')
            #     if count == 30000:
            #         print('上报结束')
            #         break
            # except Exception as e:
            #     pass
            res_key = message.key.decode()
            try:
                if 202300001 <= int(res_key) < 202330001:
                    count += 1
                if count == 1:
                    time_start = datetime.fromtimestamp(message.timestamp / 1000)
                    print(f'第一条kafka消息入库的时间为 {time_start}')
                    tmp_list.append(time_start)
                if count % 500 == 0 and count > 1:
                    time_end = datetime.fromtimestamp(message.timestamp / 1000)
                    tmp_list.append(time_end)
                    # time_res = time_end - time_start
                    # print(f'增加了{count}个，用时{time_res}秒')
                if count == 30000:
                    print(tmp_list)
                    print(tmp_list[-1] - tmp_list[0])
                    break
            except Exception as e:
                print(e)


if __name__ == '__main__':
    # ok = OperateKafka(version=156)
    # # ok.send_message('EVENT_PUSH_SERVICE_EVENT_PUSH', data)
    # fmt_s = '%Y%m%d%H%M%S'
    # count = 0
    # for j in range(0 + 1, 0 + 11):
    #     # for j in range(96):
    #     # device_id = str(100000000000000000 + j)
    #     device_id = str(10000000 + j)
    #     # device_id = '121926203731877888'
    #     tm = datetime(2022, 10, 1, 0, 0)
    #     for i in range(2880):
    #         tm_res = str(tm.strftime(fmt_s))[::-1]
    #         row_key = tm_res + device_id + 'mi15'
    #         data = {"indexColumns": "device_id,data_time,frequency", "rowkey": row_key,
    #                 "tableName": "ods_device_electric_ts"}
    #         ok.send_message('kafka.index.flink', data)
    #         tm += timedelta(minutes=15)
    #         count += 1
    #         print(count)
    # # ok.send_message('COMMUNICATION_RESPONSE_ANALOG', data)
    ok = OperateKafka(version='dxy')
    # data = [
    #     {
    #         "dataSource": 1,
    #         "responseDeviceIdsMap": {},
    #         "responseId": "201500001-1653444011883",
    #         "responseResults": [
    #             {
    #                 "businessCode": "DeviceInitVerchg",
    #                 "businessType": "event_electric",
    #                 "deviceResults": [
    #                     {
    #                         "code": 20000,
    #                         "data": [
    #                             {
    #                                 "ver_before": "5678",
    #                                 "data_time": 1427286720000,
    #                                 "event_source": "0",
    #                                 "ver_after": "9@AB"
    #                             }
    #                         ],
    #                         "deviceCode": "920204059",
    #                         "deviceId": "658211",
    #                         "frequency": "mi15",
    #                         "regionCode": "320115",
    #                         "success": True,
    #                         "sysType": [
    #                             "es"
    #                         ]
    #                     }
    #                 ],
    #                 "schemaCode": "QKL8163_2_2018",
    #                 "serviceIdentifier": "0EH"
    #             },
    #             {
    #                 "businessCode": "deivce_version_change",
    #                 "businessType": "param",
    #                 "deviceResults": [
    #                     {
    #                         "code": 20000,
    #                         "data": [
    #                             {
    #                                 "vh": "9@AB",
    #                                 "qv": "5678",
    #                                 "sor": "0",
    #                                 "eventtm": 1427286720000,
    #                                 "cshbz": "1",
    #                                 "mk1bz": "0",
    #                                 "type": "1",
    #                                 "mk2bz": "0"
    #                             }
    #                         ],
    #                         "deviceCode": "920204059",
    #                         "deviceId": "658211",
    #                         "frequency": "mi15",
    #                         "regionCode": "320115",
    #                         "success": True,
    #                         "sysType": [
    #                             "es"
    #                         ]
    #                     }
    #                 ],
    #                 "schemaCode": "QKL8163_2_2018",
    #                 "serviceIdentifier": "0EH"
    #             }
    #         ],
    #         "responseTime": "2022-05-25 10:00:11",
    #         "responseType": 2,
    #         "sysCode": "tcp"
    #     }
    # ]
    # data = [
    #     {
    #         "up_fs_val": -12,
    #         "down_fs_val": -12,
    #         "up_fs_signal_ratio": -1,
    #         "down_fs_signal_ratio": -3.5,
    #         "ua_origin_v": 100.5,
    #         "pc_actual_kw": 0.0679,
    #         "uub_origin_d": "",
    #         "dfap_origin_kw": "",
    #         "indrrt_origin_kvarh": 4.68,
    #         "real_time": 1653459300000,
    #         "pfc_origin_empty": 0.9699,
    #         "efab_actual_kwh": 0.0,
    #         "frequency": "mi15",
    #         "tableName": "ods_device_electric_ts",
    #         "indfab_origin_kwh": 0.0,
    #         "indraa_origin_kwh": 35.4253,
    #         "dfap_time_empty": "",
    #         "dfas_time_empty": "",
    #         "data_time": 1653459300000,
    #         "erab_actual_kwh": 0.0,
    #         "dfas_origin_kw": "",
    #         "pfa_origin_empty": 0.7206,
    #         "paa_origin_kw": 0.0408,
    #         "indrab_origin_kwh": 0.0,
    #         "device_id": "184671893815566336",
    #         "uac_origin_d": 59.7,
    #         "tc_origin_dc": 23.1,
    #         "dfaf_origin_kw": "",
    #         "uc_origin_v": 100.5,
    #         "indfrt_origin_kvarh": 465.81,
    #         "indfaa_origin_kwh": 855.984,
    #         "qc_actual_kvar": -0.0171,
    #         "data_source": "1",
    #         "iaa_origin_d": 316.1,
    #         "indrac_origin_kwh": 26.4653,
    #         "iz_origin_a": "",
    #         "ub_origin_v": 0.0,
    #         "efaa_actual_kwh": 0.0082,
    #         "pb_actual_kw": 0.0,
    #         "dfat_origin_kw": "",
    #         "dfaf_time_empty": "",
    #         "data_qa": "",
    #         "dfao_time_empty": "",
    #         "erat_actual_kwh": 0.0,
    #         "indfac_origin_kwh": 1133.5509,
    #         "ic_origin_a": 0.697,
    #         "erac_actual_kwh": 0.0,
    #         "efrt_actual_kvarh": 0.01,
    #         "pa_actual_kw": 0.0469,
    #         "pac_origin_kw": 0.0626,
    #         "tb_origin_dc": 23.1,
    #         "indfat_origin_kwh": 1989.5349,
    #         "efac_actual_kwh": 0.0135,
    #         "dfat_maxtime_empty": "",
    #         "update_time": 1653459300000,
    #         "dt_origin_kw": 0.0867,
    #         "pfb_origin_empty": 0.9999,
    #         "qb_actual_kvar": 0.0,
    #         "indrat_origin_kwh": 61.8906,
    #         "iab_origin_d": 0.0,
    #         "uab_origin_d": 0.0,
    #         "qa_actual_kvar": 0.0451,
    #         "pft_origin_empty": 0.9,
    #         "ta_origin_dc": 23.1,
    #         "pt_actual_kw": 0.1147,
    #         "iub_origin_d": "",
    #         "eraa_actual_kwh": 0.0,
    #         "indrp1_origin_kvarh": 450.21,
    #         "dfao_origin_kw": "",
    #         "indrp3_origin_kvarh": 0.85,
    #         "indrp4_origin_kvarh": 3.83,
    #         "indrp2_origin_kvarh": 15.6,
    #         "ib_origin_a": 0.0,
    #         "errt_actual_kvarh": 0.0,
    #         "qt_actual_kvar": 0.028,
    #         "efat_actual_kwh": 0.0217,
    #         "uaa_origin_d": 0.0,
    #         "ia_origin_a": 0.647,
    #         "pab_origin_kw": 0.0,
    #         "iac_origin_d": 73.8
    #     }
    # ]
    # ok.send_message(topic='COMMUNICATION_RESPONSE_DATA_PROCESSED', message=data)
    ok.consumer_demo_2('COMMUNICATION_RESPONSE_DATA')
